// Top10Reducer.java
package com.taxi.analysis.regionTop10;

import java.io.IOException;
import java.util.TreeMap;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class Top10Reducer extends Reducer<NullWritable, Text, Text, Text> {

    private TreeMap<Integer, String> finalTop10 = new TreeMap<>();

    @Override
    public void reduce(NullWritable key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        for (Text value : values) {
            try {
                String line = value.toString();
                String[] parts = line.split("	");

                if (parts.length == 2) {
                    String region = parts[0];
                    int count = Integer.parseInt(parts[1]);

                    finalTop10.put(count, region);

                    // 保持只有前10个
                    if (finalTop10.size() > 10) {
                        finalTop10.remove(finalTop10.firstKey());
                    }
                }
            } catch (Exception e) {
                System.err.println("Error in reducer: " + value.toString());
            }
        }

        // 输出最终TOP10结果
        int rank = 1;
        for (int count : finalTop10.descendingKeySet()) {
            String region = finalTop10.get(count);
            context.write(new Text("Rank_" + rank),
                    new Text(region + "	" + count + "	接单数"));
            rank++;
        }
    }
}
